package org.redis.service.impl;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redis.constance.Constance;
import org.redis.service.RedisKeys;
import org.redis.service.RedisService;
import org.redis.utils.ThreadPool;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * redis 服务类
 *
 * @author admin
 */
@Slf4j
@Service
@RequiredArgsConstructor
@SuppressWarnings("unchecked")
public class RedisServiceImpl implements RedisService {

    private final RedisTemplate<String, String> redisTemplate;

    private final RedissonClient redisson;

    @Override
    public <T extends RedisKeys, R> R getAndLoader(T key, Function<T, R> t) {
        ValueOperations<String, String> operations = redisTemplate.opsForValue();
        R value = (R) operations.get(key.getKey());
        if (value == null) {
            return computeAbsent(key, t, true);
        }
        return value;
    }

    @Override
    public <T extends RedisKeys, R> R getAndAsyncLoader(T key, Function<T, R> t) {
        ValueOperations<String, String> operations = redisTemplate.opsForValue();
        R value = (R) operations.get(key.getKey());
        if (value == null) {
            return computeAbsent(key, t, false);
        }
        return value;
    }
    private <T extends RedisKeys, R> R computeAbsent(T key, Function<T, R> mappingFunction, Boolean flag) {
        Objects.requireNonNull(mappingFunction);
        ValueOperations<String, String> operations = redisTemplate.opsForValue();
        RLock lock = redisson.getLock( key.getKey()+ Constance.LOCK_SUFFIX);
        try {
            lock.lock();
            R data = (R) operations.get(key.getKey());
            if (data == null) {
                if (flag) {
                    data = syncResult(key, mappingFunction);
                    log.trace("数据同步初始化成功 key：{} data{}", key.getKey(), data);
                } else {
                    asyncResult(key, mappingFunction);
                    log.trace("数据异步初始化成功 key：{} ", key.getKey());
                }
            }
            log.trace("被加锁的线程： {} ----------", Thread.currentThread().getName());
            return data;
        } catch (InterruptedException | ExecutionException e) {
            log.error("异步执行获取数据异常 ", e);
            throw new IllegalStateException(e);
        } finally {
            lock.unlock();
        }
    }

    private <T extends RedisKeys, R> R syncResult(T key, Function<T, R> mappingFunction) throws InterruptedException, ExecutionException {
        R data = supplyAsync(key, mappingFunction).get();
        syncRedis(data, key);
        return data;
    }
    private <T extends RedisKeys, R> void asyncResult(T key, Function<T, R> mappingFunction) {
        CompletableFuture.runAsync(() -> {
            try {
                syncResult(key, mappingFunction);
            } catch (ExecutionException | InterruptedException e) {
                log.error("异步加载数据执行失败!", e);
            }
        });
    }

    private <T extends RedisKeys, R> void syncRedis(R data, T key) {
        if (data != null) {
            ValueOperations<String, String> operations = redisTemplate.opsForValue();
            if (key.getTime().equals(-1L)) {
                operations.set(key.getKey(), (String) data);
            } else {
                operations.set(key.getKey(), (String) data, key.getTime(), TimeUnit.SECONDS);
            }
        }
    }

    private <T extends RedisKeys, R> CompletableFuture<R> supplyAsync(T key, Function<T, R> mappingFunction) {
        return CompletableFuture.supplyAsync(() -> mappingFunction.apply(key), ThreadPool.getExecutor());
    }

    @Override
    public <T extends RedisKeys> void addList(T keys, List<Integer> data, Integer length) {
        String key = keys.getKey();
        redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                ListOperations listOperations = operations.opsForList();
                if (length > 0 && data.size() > length) {
                    List<Integer> list = data.subList(0, length);
                    listOperations.rightPushAll(key, list.stream().map(String::valueOf).toArray());
                } else {
                    listOperations.rightPushAll(key, data.stream().map(String::valueOf).toArray());
                }
                if (length > 0) {
                    listOperations.trim(key, 0, length);
                }
                return null;
            }
        });
    }

    @Override
    public <T extends RedisKeys> void addList(T keys) {
        redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                ListOperations<String, String> listOperations = operations.opsForList();
                listOperations.remove(keys.getKey(), 1L, String.valueOf(keys.getValue()));
                listOperations.leftPush(keys.getKey(), String.valueOf(keys.getValue()));
                return null;
            }
        });
    }

    @Override
    public <T extends RedisKeys> void deleteList(T keys) {
        redisTemplate.opsForList().remove(keys.getKey(), 1L, String.valueOf(keys.getValue()));
    }

    @Override
    public <T extends RedisKeys, V extends Collection<Integer>> void addSet(T keys, V data) {
        redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                SetOperations set = redisTemplate.opsForSet();
                if (!CollectionUtils.isEmpty(data)) {
                    set.add(keys.getKey(), data.stream().map(String::valueOf).toArray());
                }
                return null;
            }
        });
    }

    @Override
    public <T extends RedisKeys, V extends Collection<Integer>> void addSetClear(T keys, V data, Long length) {
        SetOperations set = redisTemplate.opsForSet();
        Long size = set.size(keys.getKey());
        if (size > length) {
            redisTemplate.delete(keys.getKey());
        }
        if (!CollectionUtils.isEmpty(data)) {
            set.add(keys.getKey(), data.stream().map(String::valueOf).toArray());
        }
    }

    @Override
    public <T extends RedisKeys> Long differenceAndStore(T key1, T key2, T store) {
        redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                SetOperations operation = operations.opsForSet();
                operation.differenceAndStore(key1.getKey(), key2.getKey(), store.getKey());
                operations.expire(store.getKey(), store.getTime(), TimeUnit.SECONDS);
                return null;
            }
        });
        return 1L;
    }

    @Override
    public <T extends RedisKeys> Set<Integer> difference(T key1, T key2) {
        SetOperations<String, String> operations = redisTemplate.opsForSet();
        Set<String> difference = operations.difference(key1.getKey(), key2.getKey());
        if (!CollectionUtils.isEmpty(difference)) {
            return difference.stream().map(Integer::parseInt).collect(Collectors.toSet());
        }
        return new HashSet<>();
    }

    @Override
    public <T extends RedisKeys> Set<Integer> randomMembers(T key, Integer count) {
        SetOperations<String, String> operations = redisTemplate.opsForSet();
        Set<String> difference = operations.distinctRandomMembers(key.getKey(), Long.valueOf(count));
        if (!CollectionUtils.isEmpty(difference)) {
            return difference.stream().map(Integer::parseInt).collect(Collectors.toSet());
        }
        return new HashSet<>();
    }

    @Override
    public <T extends RedisKeys> Set<Integer> getAllMembers(T key) {
        SetOperations<String, String> operations = redisTemplate.opsForSet();
        Set<String> difference = operations.members(key.getKey());
        if (!CollectionUtils.isEmpty(difference)) {
            return difference.stream().map(Integer::parseInt).collect(Collectors.toSet());
        }
        return new HashSet<>();

    }

    @Override
    public <T extends RedisKeys> void removeMembers(T key) {
        redisTemplate.opsForSet().remove(key.getKey(), String.valueOf(key.getValue()));
    }

    @Override
    public <T extends RedisKeys> void addMembers(T key) {
        redisTemplate.opsForSet().add(key.getKey(), String.valueOf(key.getValue()));
    }

    @Override
    public <T extends RedisKeys> List<Integer> getList(T key, Long start, Long end) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        List<String> range = listOperations.range(key.getKey(), start, end);
        if (!CollectionUtils.isEmpty(range)) {
            return range.stream().map(Integer::parseInt).collect(Collectors.toList());
        }
        return new ArrayList<>();
    }

    @Override
    public <T extends RedisKeys> void deleteKey(T key) {
        redisTemplate.delete(key.getKey());
    }

    @Override
    public <T extends RedisKeys> String get(T key) {
        return redisTemplate.opsForValue().get(key.getKey());
    }

    @Override
    public <T extends RedisKeys,R> R get(T key,Class<R> type) {
        String redisData = redisTemplate.opsForValue().get(key.getKey());
        if(StringUtils.isNotBlank(redisData)){
            return convert(redisData, type);
        }
        return null;
    }

    @Override
    public <T extends RedisKeys> Boolean ifAbsent(T key) {
        Long time = key.getTime();
        Boolean flag = null;
        if(time != null && !time.equals(-1L)){
            flag = redisTemplate.opsForValue().setIfAbsent(key.getKey(),String.valueOf(key.getValue()),key.getTime(),TimeUnit.SECONDS);
        }else {
            flag = redisTemplate.opsForValue().setIfAbsent(key.getKey(),String.valueOf(key.getValue()));
        }
        return Boolean.TRUE.equals(flag);
    }

    @Override
    public <T extends RedisKeys> void add(T key) {
        Long time = key.getTime();
        if(time != null && !time.equals(-1L)){
            redisTemplate.opsForValue().set(key.getKey(),String.valueOf(key.getValue()),key.getTime(),TimeUnit.SECONDS);
        }else {
            redisTemplate.opsForValue().set(key.getKey(),String.valueOf(key.getValue()));
        }
    }

    @Override
    public <T extends RedisKeys> void expire(T key) {
        Long time = key.getTime();
        if(time != null && !time.equals(-1L)){
            redisTemplate.expire(key.getKey(),key.getTime(),TimeUnit.SECONDS);
        }
    }

    @Override
    public <T extends RedisKeys> Long addInc(T key) {
        Long time = key.getTime();
        if(time != null && !time.equals(-1L)){
            redisTemplate.opsForValue().setIfAbsent(key.getKey(),String.valueOf(key.getValue()),time,TimeUnit.SECONDS);
        }else{
            redisTemplate.opsForValue().setIfAbsent(key.getKey(),String.valueOf(key.getValue()));
        }
        return redisTemplate.opsForValue().increment(key.getKey());
    }

}